package com.shujia.flink.sink;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

public class Demo2MySInk {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //使用自定义sink
        linesDS.addSink(new MySink());

        env.execute();
    }
}

//自定义sink
class MySink implements SinkFunction<String> {
    @Override
    public void invoke(String value, Context context) throws Exception {
        //自定义数据sink的位置
        System.out.println("自定义sink:" + value);
    }
}
